//
// Created by yanhai on 2021/11/5.
//

#include <iostream>
#include <thread>

#include "utility/hashthread/hash_table_thread.h"
#include "utility/hashthread/spare_queue_proxy.h"

#include "flow.h"
#include "flow_util.h"
#include "packet.h"

static SpareQueueProxy *spareQueueProxy;

static Flow *FlowGetNew()
{
    Flow *f = reinterpret_cast<Flow *>(spareQueueProxy->Dequeue());
    if (f == nullptr) {
        f = FlowAlloc();
        if (f == nullptr)
            return nullptr;
    }

    f->mutex.lock();
    return f;
}

static void FlowToQueue(HashTableThread::node_type f)
{
    spareQueueProxy->Enqueue(f);
}

void worker_thread(HashTableThread *hash)
{
    unsigned int delay = 10000;

    srand(time(nullptr));

    while (delay--) {
        time_t t = time(nullptr);
        unsigned int port = rand() % 1000;
        Packet *p = PacketAlloc(1, 2, 3, port, t);
        auto node = hash->Find(p);
//        auto f = dynamic_cast<Flow *>(node);
        // 为了提高性能，没有使用 dynamic_cast
        auto f = reinterpret_cast<Flow *>(node);
        assert(f != nullptr);
        f->count++;
        f->last_time = t;
//        std::cout << "flow count: " << f->count << std::endl;
        f->mutex.unlock();
        PacketFree(p);
        std::this_thread::sleep_for(std::chrono::milliseconds (10));
    }
    std::cout << "worker thread exit" << std::endl;
}

void flow_manager(HashTableThread *hash)
{
    // 检测flow超时
    auto timeout_checker = [](HashTableThread::node_type node, std::time_t now) {
        auto f = reinterpret_cast<Flow *>(node);
        return (now - f->last_time) > 30;
    };

    auto callback = [](HashTableThread::node_type node) {
        auto f = reinterpret_cast<Flow *>(node);
        std::cout << "flow timeout: "
                  << f->src_addr << ":" << f->src_port << " -> "
                  << f->dst_addr << ":" << f->dst_port
                  << ", count=" << f->count << std::endl;
    };

    while (true) {
        time_t t = time(nullptr);
        spareQueueProxy->Update();
        std::cout << "spare queue: " << spareQueueProxy->Len() << std::endl;
        hash->CheckTimeout(t, timeout_checker, callback);
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}

int main()
{
    auto *hash = new HashTableThread(1024, FlowGetNew, FlowToQueue);
    spareQueueProxy = new SpareQueueProxy(500, FlowAlloc,
                                          [](SpareQueueProxy::value_type *v) {
                                              auto f = reinterpret_cast<Flow *>(v);
                                              FlowFree(f);
                                          });
    spareQueueProxy->Prealloc();

    std::thread t1(worker_thread, hash);
    std::thread t3(worker_thread, hash);
    std::thread t2(flow_manager, hash);

    t1.join();
    t3.join();
    t2.join();

    return 0;
}
